package com.xiehua.task.persistence.rocks;

import com.xiehua.task.prop.XiehuaConfig;
import com.xiehua.task.utils.StringUtils;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectFactory;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.rocksdb.*;
import org.rocksdb.util.SizeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.time.LocalDate;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.xiehua.task.prop.XiehuaConfig.PERSISTENCE_PATH;

public class RocksFactory implements PooledObjectFactory<XeihuaPersitence> {

    private static final Logger logger = LoggerFactory.getLogger(RocksFactory.class);

    public static final String COLUMN_FAMILY_TASK_SUCCESS = "task_success";

    public static final String COLUMN_FAMILY_TASK_FAIL = "task_fail";

    static {
        RocksDB.loadLibrary();
    }

    @Override
    public PooledObject<XeihuaPersitence> makeObject() throws Exception {
        logger.info("初始化RocksDB...");
        //init Options
        List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
        //init Options
        Options options = new Options();
        Statistics stats = new Statistics();
        options.setCreateIfMissing(true)
                .setStatistics(stats)
                .setWriteBufferSize(8 * SizeUnit.KB)
                .setMaxWriteBufferNumber(3)
                .setMaxBackgroundCompactions(10)
                .setCompressionType(CompressionType.SNAPPY_COMPRESSION)
                .setCompactionStyle(CompactionStyle.UNIVERSAL);
        Map<String, String> configMap = XiehuaConfig.getInstance();
        String dbPath = RocksPersitence.checkPersitencePath(configMap);
        List<byte[]> cfs = RocksDB.listColumnFamilies(options, dbPath);
        if (cfs.size() > 0) {
            for (byte[] cf : cfs) {
                columnFamilyDescriptors.add(new ColumnFamilyDescriptor(cf, new ColumnFamilyOptions()));
            }
        } else {
            columnFamilyDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY, new ColumnFamilyOptions()));
        }

        List<ColumnFamilyHandle> handles = new ArrayList<>();
        DBOptions dbOptions = new DBOptions();
        dbOptions.setCreateIfMissing(true);
        RocksDB rocksDB = RocksDB.open(dbOptions, dbPath, columnFamilyDescriptors, handles);
        //check ColumnFamilyHandle
        Map<String, ColumnFamilyHandle> map = handles.stream().collect(Collectors.toMap(s -> new String(s.getName()), t -> t, (x, y) -> y));
        if (!map.containsKey(COLUMN_FAMILY_TASK_SUCCESS))
            map.put(COLUMN_FAMILY_TASK_SUCCESS, rocksDB.createColumnFamily(new ColumnFamilyDescriptor(COLUMN_FAMILY_TASK_SUCCESS.getBytes(), new ColumnFamilyOptions())));
        if (!map.containsKey(COLUMN_FAMILY_TASK_FAIL))
            map.put(COLUMN_FAMILY_TASK_FAIL, rocksDB.createColumnFamily(new ColumnFamilyDescriptor(COLUMN_FAMILY_TASK_FAIL.getBytes(), new ColumnFamilyOptions())));

        return new DefaultPooledObject<>(new XeihuaPersitence(rocksDB, map));
    }

    @Override
    public void destroyObject(PooledObject<XeihuaPersitence> p) throws Exception {
        XeihuaPersitence rocksDB = p.getObject();
        if (rocksDB != null) {
            logger.info("关闭RocksDB连接");
            rocksDB.close();
        }
        ;
    }

    @Override
    public boolean validateObject(PooledObject<XeihuaPersitence> p) {
        return false;
    }

    @Override
    public void activateObject(PooledObject<XeihuaPersitence> p) throws Exception {
        logger.info("RocksDB对象重置...");
    }

    @Override
    public void passivateObject(PooledObject<XeihuaPersitence> p) throws Exception {
        logger.info("RocksDB对象使用完毕...");
    }

}
